/*-------------------------------------------------------------------------
 *
 * worker_internal.h
 *	  逻辑复制工作者共享的内部头文件。
 *
 * Portions Copyright (c) 2016-2022, PostgreSQL Global Development Group
 *
 * src/include/replication/worker_internal.h
 *
 *-------------------------------------------------------------------------
 */
#ifndef WORKER_INTERNAL_H
#define WORKER_INTERNAL_H

#include <signal.h>

#include "access/xlogdefs.h"
#include "catalog/pg_subscription.h"
#include "datatype/timestamp.h"
#include "storage/fileset.h"
#include "storage/lock.h"
#include "storage/spin.h"
#ifdef FDDM
#include "replication/logicalrelation.h"
#include "executor/execPartition.h"
#endif


typedef struct LogicalRepWorker
{
	/* 此工作线程启动的时间。 */
	TimestampTz launch_time;

	/* 指示此槽是使用中还是空闲。 */
	bool		in_use;

	/* 每当新工作线程占用此槽时增加。 */
	uint16		generation;

	/* 指向进程数组的指针。如果未运行则为NULL。 */
	PGPROC	   *proc;

	/* 连接的数据库ID。 */
	Oid			dbid;

	/* 用于连接的用户（将与订阅的所有者相同）。 */
	Oid			userid;

	/* 工作线程的订阅ID。 */
	Oid			subid;

	/* 用于初始表同步。 */
	Oid			relid;
	char		relstate;
	XLogRecPtr	relstate_lsn;
	slock_t		relmutex;

	/*
	 * 用于创建流式事务的更改和子事务文件。
	 * 在第一个流式事务到达时，将初始化文件集，并在工作线程退出时删除。
	 * 在此情况下，将为每个事务创建单独的缓冲文件，
	 * 这些文件将在事务完成后删除。
	 */
	FileSet    *stream_fileset;

	/* Stats. */
	XLogRecPtr	last_lsn;
	TimestampTz last_send_time;
	TimestampTz last_recv_time;
	XLogRecPtr	reply_lsn;
	TimestampTz reply_time;
#ifdef FDDM
	bool 		forward_subs_change;
#endif
} LogicalRepWorker;

#ifdef FDDM
typedef struct ApplyExecutionData ApplyExecutionData;

typedef enum FddHandleType
{
	Fdd_Insert,
	Fdd_Update,
	Fdd_Delete,
	Fdd_Update_Part,
	Fdd_Begin,
	FDD_Commit
} FddHandleType;

typedef struct ApplyExecState
{
	EState			   *estate;
	EPQState		   *epqstate;
	ResultRelInfo	   *resultRelInfo;
	TupleTableSlot	   *slot;
	LogicalRepRelMapEntry *targetRel;
	LogicalRepTupleData *newtup;
	
	//用于分区表update
	ResultRelInfo         *partrelinfo;
	ApplyExecutionData    *edata;
	TupleTableSlot        *remoteslot_part;
	TupleConversionMap    *map;
} ApplyExecState;


typedef bool (*sub_is_normal_hook_type)(Oid subid, const char *subname);
typedef bool (*handle_conflict_hook_type)(FddHandleType type, ApplyExecState *apestate);

typedef List *(*get_forward_subs_hook_type)(Oid sub_id);

//重写ExecFindPartition函数，主要用于处理分区表确实场景
typedef ResultRelInfo *(*find_partition_info_hook_type)(ModifyTableState *mtstate,
				  ResultRelInfo *rootResultRelInfo,
				  PartitionTupleRouting *proute,
				  TupleTableSlot *slot, 
				  EState *estate,
				  LogicalRepRelation *remoterel, 
				  LogicalRepTupleData *tup);
#endif




/* 应用工作线程的主内存上下文。在工作线程的生命周期内是永久的。 */
extern PGDLLIMPORT MemoryContext ApplyContext;

/* libpqreceiver连接 */
extern PGDLLIMPORT struct WalReceiverConn *LogRepWorkerWalRcvConn;

/* 工作线程和订阅对象。 */
extern PGDLLIMPORT Subscription *MySubscription;
extern PGDLLIMPORT LogicalRepWorker *MyLogicalRepWorker;

extern PGDLLIMPORT bool in_remote_transaction;

extern void logicalrep_worker_attach(int slot);
extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
												bool only_running);
extern List *logicalrep_workers_find(Oid subid, bool only_running);
extern void logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
									 Oid userid, Oid relid);
extern void logicalrep_worker_stop(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);

extern int	logicalrep_sync_worker_count(Oid subid);

extern void ReplicationOriginNameForTablesync(Oid suboid, Oid relid,
											  char *originname, int szorgname);
extern char *LogicalRepSyncTableStart(XLogRecPtr *origin_startpos);

extern bool AllTablesyncsReady(void);
extern void UpdateTwoPhaseState(Oid suboid, char new_state);

extern void process_syncing_tables(XLogRecPtr current_lsn);
extern void invalidate_syncing_table_states(Datum arg, int cacheid,
											uint32 hashvalue);

static inline bool
am_tablesync_worker(void)
{
	return OidIsValid(MyLogicalRepWorker->relid);
}

#endif							/* WORKER_INTERNAL_H */
